#ifndef _PIPEEP_H_
#define _PIPEEP_H_

#define _CRT_SECURE_NO_DEPRECATE

enum {
	PIPE_EP_START_SRC,
	PIPE_EP_START_DST,
	PIPE_WRITE_STREAM,
	PIPE_WRITE_SOCKET,
	PIPE_EP_SHUTDOWN,
};

class MyPktSocket;

#define CONNECT_INTERVAL (20000)

class PipeEp : public talk_base::MessageHandler, public sigslot::has_slots<>
{
public:
	PipeEp(ConnectionCallback* cc,
	           PipeClient* pump,
		   talk_base::Thread *th,
		   //talk_base::SocketServer *ss,
		   bool src,
          	   std::string& user,
		   std::string& desc,
		   std::string& max_packet_size
	       );

	virtual ~PipeEp(void);

	void StartSrc();


	void InitiateTunnel();

	void ProcessStream();

	void OnStreamEvent(talk_base::StreamInterface* stream, int events, int error);

	void OnMessage(talk_base::Message *pmsg);

	bool DstConnect();

	void OnIncomingTunnel(cricket::TunnelSessionClient* client, buzz::Jid jid,
		std::string description, cricket::Session* session);

	void OnPacket(const char* buf, size_t size, const talk_base::SocketAddress& remote_addr,
		talk_base::AsyncPacketSocket* socket);

	void Show(void);

	void ConnectTimeout(void);

        int WriteToStream(const char* buf, size_t size);

private:


	void WriteToSocket();

	void OnSocketClose(talk_base::AsyncSocket *);
	void CleanupAll();
        void Shutdown();

        ConnectionCallback *connection_callback_;
	PipeClient *pump_;
	talk_base::Thread *thread_;
	//talk_base::SocketServer *ss_;


	bool issrc_;
	std::string user_;
	std::string desc_;
	std::string message_size_str_;
	std::string rportstr_;
	std::string description_;
	int message_size_;

	uint32 first_msg_byte_sent_at_;
	uint32 last_msg_byte_sent_at_;

	int message_number_;
	int read_number_;

	uint32 first_msg_byte_received_at_;
	uint32 last_msg_byte_received_at_;

	uint32 avg_write_time_;
	uint32 avg_read_time_;
	int32 avg_jitter_;
	int32 avg_reads_per_message_;
	uint32 avg_network_time_;
	uint32 bytes_at_last_print_;
	uint32 messages_at_last_print_;
	uint32 time_at_last_print_;

	//int lport_,rport_;
	talk_base::AsyncSocket *socket_;      //server socket ep
	talk_base::StreamInterface *stream_;  //tunnel stream interface
	cricket::TunnelSessionClient* client_;
	cricket::Session *session_;
	char read_buf_[1024*64];
	size_t read_buf_len_;
	char write_buf_[1024*64];
	size_t write_buf_len_;
	int pending_stream_read_;
	// stats
	size_t stream_bytes_,socket_bytes_;
	uint32 start_time_;
	// for udp
    //char test_data_buf_[1024*1024];
};

#endif
